package org.fastsyncer.manager.framework;

import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.entity.DataSourceConfig;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.connector.common.DatabaseConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TaskDistributor {

	private static Logger logger = LoggerFactory.getLogger(TaskDistributor.class);

	private static TaskDistributor instance = null;

	private TaskDistributor() {
	}

	public static TaskDistributor getInstance() {
		if (instance == null) {
			synchronized (TaskDistributor.class) {
				instance = new TaskDistributor();
			}
		}
		return instance;
	}

	/**
	 * 计算总进度
	 * @param mappingTask
	 * @return
	 */
	public void countProcess(MappingTask mappingTask) {
		// 总数据量
		int total = mappingTask.getSourceMapping().getRowCount();
		// 完成进度(必须大于等于0)
		int completed = mappingTask.getProcessCompleted();
		// 任务上线值，默认10W，建议大于batchNum
		int maxTask = mappingTask.getMaxTask();
		// 总进度
		int process = total % maxTask == 0 ? total / maxTask : total / maxTask + 1;
		mappingTask.setProcess(process);
		logger.info("数据源总量:" + total);
		logger.info("总进度:" + process);
		logger.info("已完成进度:" + completed);
	}
	
	/**
	 * 计算当前进度任务
	 * @param mappingTask
	 * @return
	 */
	public ConcurrentLinkedQueue<Integer[]> distributeTask(MappingTask mappingTask) {
		Mapping sourceMapping = mappingTask.getSourceMapping();
		ConcurrentLinkedQueue<Integer[]> distribute = null;
		//暂时只做关系型任务拆分算法
		switch (sourceMapping.getConnector()) {
		case ConnectorConstant.DB_CONNECTOR:
			DataSourceConfig cfg = (DataSourceConfig) sourceMapping.getConfig();
			if(StringUtils.equalsIgnoreCase(DatabaseConstant.DB_MYSQL_DRIVER_CLASSNAME, cfg.getDriverClassName())){
				distribute= distributeConnectorMysql(mappingTask);
				break;
			}
			
			if(StringUtils.equalsIgnoreCase(DatabaseConstant.DB_ORACLE_DRIVER_CLASSNAME, cfg.getDriverClassName())){
				distribute= distributeConnectorOracle(mappingTask);
				break;
			}
			break;
		default:
			break;
		}
		return distribute;
	}
	
	/**
	 * 计算Mysql当前进度任务
	 * @param mappingTask
	 * @return
	 */
	public ConcurrentLinkedQueue<Integer[]> distributeConnectorMysql(MappingTask mappingTask) {
		logger.info("计算拆分任务");
		// 总数据量
		int total = mappingTask.getSourceMapping().getRowCount();
		// 完成进度(必须大于等于0)
		int completed = mappingTask.getProcessCompleted();
		// 任务上线值，默认10W，建议大于batchNum
		int maxTask = mappingTask.getMaxTask();
		// 每次批量执行数(必须大于0)
		int batchNum = mappingTask.getBatchNum();
		logger.info("数据源总量:" + total);
		logger.info("已完成进度:" + completed);

		// 计算本次任务总量
		int t_start = completed * maxTask;
		int t_end = t_start + maxTask;
		t_end = t_end >= total ? total : t_end;
		// 所有小任务总量
		int count = t_end - t_start;
		logger.info("上限同步数:" + maxTask);
		logger.info("当前总数:" + count);
		logger.info("批量执行数:" + batchNum);

		// 计算小任务量
		int taskNum = count % batchNum == 0 ? count / batchNum : count / batchNum + 1;
		// 拆分小任务
		ConcurrentLinkedQueue<Integer[]> quene = new ConcurrentLinkedQueue<>();
		for (int i = 0; i < taskNum; i++) {
			// 从任务完成进度起点位置开始计算
			int start = t_start + i * batchNum;
			// 判断是否已到最后
			int end = start + batchNum > t_end ? t_end - start : batchNum;
			quene.offer(new Integer[] { start, end });
		}
		logger.info("共计任务数：" + taskNum);
		return quene;
	}
	
	/**
	 * 计算Mysql当前进度任务
	 * @param mappingTask
	 * @return
	 */
	public ConcurrentLinkedQueue<Integer[]> distributeConnectorOracle(MappingTask mappingTask) {
		logger.info("计算拆分任务");
		// 总数据量
		int total = mappingTask.getSourceMapping().getRowCount();
		// 完成进度(必须大于等于0)
		int completed = mappingTask.getProcessCompleted();
		// 任务上线值，默认10W，建议大于batchNum
		int maxTask = mappingTask.getMaxTask();
		// 每次批量执行数(必须大于0)
		int batchNum = mappingTask.getBatchNum();
		logger.info("数据源总量:" + total);
		logger.info("已完成进度:" + completed);

		// 计算本次任务总量
		int t_start = completed * maxTask;
		int t_end = t_start + maxTask;
		t_end = t_end >= total ? total : t_end;
		// 所有小任务总量
		int count = t_end - t_start;
		logger.info("上限同步数:" + maxTask);
		logger.info("当前总数:" + count);
		logger.info("批量执行数:" + batchNum);

		// 计算小任务量
		int taskNum = count % batchNum == 0 ? count / batchNum : count / batchNum + 1;
		// 拆分小任务
		ConcurrentLinkedQueue<Integer[]> quene = new ConcurrentLinkedQueue<>();
		for (int i = 0; i < taskNum; i++) {
			// 从任务完成进度起点位置开始计算
			int start = t_start + i * batchNum;
			// 判断是否已到最后
			int end = start + batchNum;
			end = end >= total ? total : end;
			quene.offer(new Integer[] { start, end });
		}
		logger.info("共计任务数：" + taskNum);
		return quene;
	}

}
